/**
 * Created by Administrator on 2015/9/10.
 */
import java.io.*;
import java.util.Map;
import java.util.HashMap;
import scala.Tuple2;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.SparkConf;


public class TestBuildTree {
    /**
     * 测试 DataConverter类的convert() method
     * 对应的car_test.info 为
     * { med,vhigh,high,low
     * med,vhigh,high,low
     * 3, 2,  5more, 4
     * 2,more,4
     * med,big,small
     * med,high,low
     * good,unacc,acc,vgood
     * }
     * 若为第一个则不显示，且对应编码为0.0，1.0，2.0，3.0，4.0
     *
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {

        //-Dspark.master=local
        try {

           /// String jars[] = {"E:\\desciontree\\out\\artifacts\\desciontree_jar\\desciontree.jar"};
           // SparkConf sparkConf = new SparkConf().setAppName("TestBuildTree").setMaster("spark://101.227.247.192:7077").setJars(jars);
            SparkConf sparkConf = new SparkConf().setAppName("TestBuildTree");
            JavaSparkContext sc = new JavaSparkContext(sparkConf);

            String datapath = "/bigdata/data1/test/test.csv                                                                                                                    ";
            JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(sc.sc(), datapath).toJavaRDD();
            // Split the data into training and test sets (30% held out for testing)
            JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.7, 0.3});
            JavaRDD<LabeledPoint> trainingData = splits[0];
            JavaRDD<LabeledPoint> testData = splits[1];

            // Set parameters.
            //  Empty categoricalFeaturesInfo indicates all features are continuous.
            Integer numClasses = 2;
            Map<Integer, Integer> categoricalFeaturesInfo = new HashMap<Integer, Integer>();
            String impurity = "gini";
            Integer maxDepth = 5;
            Integer maxBins = 32;

            // Train a DecisionTree model for classification.
            final DecisionTreeModel model = DecisionTree.trainClassifier(trainingData, numClasses,
                    categoricalFeaturesInfo, impurity, maxDepth, maxBins);

            // Evaluate model on test instances and compute test error
            JavaPairRDD<Double, Double> predictionAndLabel =

                    testData.mapToPair(new PairFunction<LabeledPoint, Double, Double>() {

                        public Tuple2<Double, Double> call(LabeledPoint p) {

                            return new Tuple2<Double, Double>(model.predict(p.features()), p.label());
                        }
                    });
            Double testErr =
                    1.0 * predictionAndLabel.filter(new Function<Tuple2<Double, Double>, Boolean>() {

                        public Boolean call(Tuple2<Double, Double> pl) {
                            return !pl._1().equals(pl._2());
                        }
                    }).count() / testData.count();


            System.out.println("Test Error: " + testErr);
            System.out.println("Learned classification tree model:\n" + model.toDebugString());

            // Save and load model
            model.save(sc.sc(), "myModelPath");
            DecisionTreeModel sameModel = DecisionTreeModel.load(sc.sc(), "myModelPath");

        } catch (Exception e) {
            e.printStackTrace();
        }

        // Load an

    }
}
